package com.shujia.sql

import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}

object Demo4TopN {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[4]").setAppName("app")
    conf.set("spark.sql.shuffle.partitions", "2")

    val sc = new SparkContext(conf)

    //    val sqlContext = new SQLContext(sc)
    //sqlContext  不能使用row_number     HiveContext可以使用

    /** *
      * HiveContext
      * 在本地运行会模拟一个hive元数据服务
      * 在集群中运行需要开启hive元数据服务
      *
      */
    val sqlContext = new HiveContext(sc)

    import sqlContext.implicits._

    val dianxinRDD = sc.textFile("spark/data/dianxin_data")

    /**
      * D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C,117210031795040,83401,8340104,301,20180503190539,20180503233517,20180503
      *
      *
      * /**
      * mdn	        string	用户手机号码
      * grid_id 		string	停留点所在电信内部网格号
      * city_id			string	业务发生城市id
      * county_id		string	停留点区县
      * duration		string	机主在停留点停留的时间长度（分钟）,lTime-eTime
      * grid_first_time	string	网格第一个记录位置点时间（秒级）
      * grid_last_time	string	网格最后一个记录位置点时间（秒级）
      * day_id			string	天分区
      */
      *
      */

    val dianXinData = dianxinRDD
      .filter(line => !line.split(",")(4).equals("\\N"))
      .map(line => {
        val split = line.split(",")
        val mdn = split(0)
        val grid_id = split(1)
        val city_id = split(2)
        val county_id = split(3)
        val duration = split(4).toInt
        val grid_first_time = split(5)
        val grid_last_time = split(6)
        val day_id = split(7)

        DianXin(mdn, grid_id, city_id, county_id, duration, grid_first_time, grid_last_time, day_id)
      })

    /**
      * 统计每个城市停留时间排名前10的用户
      *
      */

    //计算每个用户在每个城市总的停留时间

    dianXinData.map(dx => {
      (dx.mdn + "-" + dx.city_id, dx.duration)
    }).reduceByKey(_ + _)
      .map(kv => {
        val mdnAndCity = kv._1.split("-")
        val mdn = mdnAndCity(0)
        val city = mdnAndCity(1)

        (city, mdn + "-" + kv._2)
      }).groupByKey()
      .map(kv => {
        val city = kv._1
        //停留时间倒叙排序 取top10
        val topN = kv._2.toList.sortBy(line => -line.split("-")(1).toInt).take(10)
        (city, topN)
      }) //.foreach(println)


    dianXinData.toDF().registerTempTable("DianXin")

    sqlContext.sql(
      """
        |select * from
        |(select mdn,city_id,s,row_number() over(partition by city_id order by s desc) as rank from
        |(select mdn,city_id,sum(duration) as s from DianXin  group by mdn,city_id) as a) as b
        |where b.rank <= 10
        |
        |
      """.stripMargin)
      .show()


  }

  case class DianXin(
                      mdn: String,
                      grid_id: String,
                      city_id: String,
                      county_id: String,
                      duration: Int,
                      grid_first_time: String,
                      grid_last_time: String,
                      day_id: String
                    )

}
